#pragma once
#include "../common/mq_msg/mq_msg.pb.h"
#include "../common/mq_proto/mq_proto.pb.h"
#include "../common/consumer.hpp"
#include "../common/helper.hpp"
#include "../common/log.hpp"

#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"

#include <mutex>
#include <condition_variable>
#include <unordered_map>


namespace xzt
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using basicCommonReponsePtr = std::shared_ptr<xzt::basicCommonReponse>;
    using basicConsumeResponsePtr = std::shared_ptr<xzt::basicConsumeResponse>;
    
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
        using args_mapType = google::protobuf::Map<std::string, std::string>;

    public:
        Channel(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec)
            : _conn(conn), _codec(codec), _cid(UUIDHelper::uuid()) {}
        
        ~Channel()
        {
            basicCancel();
        }

        std::string cid() const
        {
            return _cid;
        }
    public:
        bool openChannel()
        {
            std::string rid = UUIDHelper::uuid();
            xzt::openChannelRequest request;
            request.set_rid(rid);
            request.set_cid(_cid);
            _codec->send(_conn, request);
            basicCommonReponsePtr response = waitResponse(rid);
            return response->success();
        }

        void closeChannel()
        {
            std::string rid = UUIDHelper::uuid();
            xzt::closeChannelRequest request;
            request.set_rid(rid);
            request.set_cid(_cid);
            _codec->send(_conn, request);
            waitResponse(rid);
        }

    public:    // 提供给用户使用的接口
        // 交换机相关
        bool declareExchange(const std::string &name, const ExchangeType &type,
                             bool durable, bool auto_delete, const args_mapType &args)
        {
            // 1.构建声明交换机的请求
            xzt::declareExchangeRequest request;
            const std::string rid = UUIDHelper::uuid();
            request.set_rid(rid);
            request.set_cid(_cid);
            request.set_exchange_name(name);
            request.set_exchange_type(type);
            request.set_durable(durable);
            request.set_auto_delete(auto_delete);
            args_mapType tmp_args(args);
            request.mutable_args()->swap(tmp_args);
            // 2.向服务器发送请求
            _codec->send(_conn, request);
            // 3.由于请求非阻塞，是异步的，需要等待响应
            basicCommonReponsePtr response = waitResponse(rid);
            // 4.返回
            return response->success();
        }

        void deleteExchange(const std::string &name)
        {
            // 1.构建删除交换机的请求
            xzt::deleteExchangeRequest request;
            const std::string rid = UUIDHelper::uuid();
            request.set_rid(rid);
            request.set_cid(_cid);
            request.set_exchange_name(name);
            // 2.向服务器发送请求
            _codec->send(_conn, request);
            // 3.由于请求非阻塞，是异步的，需要等待响应
            waitResponse(rid);
        }

        // 消息队列相关
        bool declareQueue(const std::string &name, bool exclusive, 
                             bool durable, bool auto_delete, const args_mapType &args)
        {
            // 1.构建声明消息队列的请求
            xzt::declareQueueRequest request;
            const std::string rid = UUIDHelper::uuid();
            request.set_rid(rid);
            request.set_cid(_cid);
            request.set_queue_name(name);
            request.set_exclusive(exclusive);
            request.set_durable(durable);
            request.set_auto_delete(auto_delete);
            args_mapType tmp_args(args);
            request.mutable_args()->swap(tmp_args);
            // 2.向服务器发送请求
            _codec->send(_conn, request);
            // 3.由于请求非阻塞，是异步的，需要等待响应
            basicCommonReponsePtr response = waitResponse(rid);
            // 4.返回
            return response->success();
        }

        void deleteQueue(const std::string &name)
        {
            // 1.构建删除消息队列的请求
            xzt::deleteQueueRequest request;
            const std::string rid = UUIDHelper::uuid();
            request.set_rid(rid);
            request.set_cid(_cid);
            request.set_queue_name(name);
            // 2.向服务器发送请求
            _codec->send(_conn, request);
            // 3.由于请求非阻塞，是异步的，需要等待响应
            waitResponse(rid);
        }

        // 绑定相关
        bool queueBind(const std::string &exchange_name, const std::string &msgqueue_name, const std::string &binding_key)
        {
            // 1.构建绑定消息队列的请求
            xzt::queueBindRequest request;
            const std::string rid = UUIDHelper::uuid();
            request.set_rid(rid);
            request.set_cid(_cid);
            request.set_exchange_name(exchange_name);
            request.set_queue_name(msgqueue_name);
            request.set_binding_key(binding_key);
            // 2.向服务器发送请求
            _codec->send(_conn, request);
            // 3.由于请求非阻塞，是异步的，需要等待响应
            basicCommonReponsePtr response = waitResponse(rid);
            // 4.返回
            return response->success();
        }

        void queueUnbind(const std::string &exchange_name, const std::string &msgqueue_name)
        {
            // 1.构建取消绑定消息队列的请求
            xzt::queueUnbindRequest request;
            const std::string rid = UUIDHelper::uuid();
            request.set_rid(rid);
            request.set_cid(_cid);
            request.set_exchange_name(exchange_name);
            request.set_queue_name(msgqueue_name);
            // 2.向服务器发送请求
            _codec->send(_conn, request);
            // 3.由于请求非阻塞，是异步的，需要等待响应
            waitResponse(rid);
        }

        // 消息发布 -- {生产者}
        void basicPublish(const std::string &exchange_name, const xzt::BasicPropeties *basic_props, const std::string &body)
        {
            // 1.构建发布消息的请求
            xzt::basicPublishRequest request;
            const std::string rid = UUIDHelper::uuid();
            request.set_rid(rid);
            request.set_cid(_cid);
            request.set_body(body);
            request.set_exchange_name(exchange_name);
            
            if(basic_props != nullptr)
            {
                request.mutable_properties()->CopyFrom(*basic_props);
            }

            // 2.向服务器发送请求
            _codec->send(_conn, request);
            // 3.由于请求非阻塞，是异步的，需要等待响应
            basicCommonReponsePtr response = waitResponse(rid);
        }

        // 消息确认
        void basicAck(const std::string &msg_id)
        {
            if(_consumer == nullptr)
            {
                ELOG("确认消息时，消费者不存在");
                return;
            }

            // 1.构建确认消息的请求
            xzt::basicAckRequest request;
            const std::string rid = UUIDHelper::uuid();
            request.set_rid(rid);
            request.set_cid(_cid);
            request.set_queue_name(_consumer->_queue_name);
            request.set_message_id(msg_id);
            // 2.向服务器发送请求
            _codec->send(_conn, request);
            // 3.由于请求非阻塞，是异步的，需要等待响应
            waitResponse(rid);
        }

        // 订阅消息队列 -- {消费者}
        bool basicConsume(const std::string& consumer_tag, const std::string& queue_name,
                         bool auto_ack, const ConsumerCallback& callback)
        {
            if(_consumer != nullptr)
            {
                ELOG("当前信道已经订阅其他队列，消费者已存在");
                return false;
            }

            // 1.构建订阅消息队列的请求
            xzt::basicConsumeRequest request;
            const std::string rid = UUIDHelper::uuid();
            request.set_rid(rid);
            request.set_cid(_cid);
            request.set_consumer_tag(consumer_tag);
            request.set_queue_name(queue_name);
            request.set_auto_ack(auto_ack);
            // 2.向服务器发送请求
            _codec->send(_conn, request);
            // 3.由于请求非阻塞，是异步的，需要等待响应
            basicCommonReponsePtr response = waitResponse(rid);
            // 4.判断请求结果
            bool ret = response->success();
            if(!ret)
            {
                ELOG("订阅消息队列失败");
                return ret;
            }
            // 5.创建消费者 (订阅成功才创建)
            _consumer = std::make_shared<Consumer>(consumer_tag, queue_name, auto_ack, callback);
            // 6.返回
            return ret;
        }
        
        // 取消订阅消息队列
        void basicCancel()
        {
            if(_consumer == nullptr)    // 说明当前信道不是消费者
            {
                return;
            }
            // 1.构建取消订阅消息队列的请求
            xzt::basicCancelRequest request;
            const std::string rid = UUIDHelper::uuid();
            request.set_rid(rid);
            request.set_cid(_cid);
            request.set_consumer_tag(_consumer->_tag);
            request.set_queue_name(_consumer->_queue_name);
            // 2.向服务器发送请求
            _codec->send(_conn, request);
            // 3.由于请求非阻塞，是异步的，需要等待响应
            waitResponse(rid);

            // 4.移除信道对应消费者身份
            _consumer.reset();
        }

    public:    // 提供给连接对象调用的接口
        // 连接收到相应时，向hash_map中添加响应
        void putOnBasicResp(const basicCommonReponsePtr& resp)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _basic_resp[resp->rid()] = resp;
            _cv.notify_all();
        }

        // 连接收到消息推送响应时，需要通过信道找到对应的消费者，利用消费者的回调函数处理消息
        bool consume(const basicConsumeResponsePtr& resp)
        {
            if(_consumer == nullptr)
            {
                ELOG("当前信道没有订阅队列，不能消费消息");
                return false;
            }
            if(_consumer->_tag != resp->consumer_tag())
            {
                ELOG("当前信道的消费者与消息响应的消费者不匹配");
                return false;
            }
            
            _consumer->_callback(resp->consumer_tag(), resp->mutable_properties(), resp->body());
            return true;
        }
    private:
        // 用于等待基础响应
        basicCommonReponsePtr waitResponse(const std::string& rid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cv.wait(lock, [this, rid]() {
                return _basic_resp.find(rid) != _basic_resp.end();
            });
            auto resp = _basic_resp[rid];
            _basic_resp.erase(rid);
            return resp;
        }
    
    private:
        std::string _cid;   // 信道id
        muduo::net::TcpConnectionPtr _conn;  // tcp连接
        ProtobufCodecPtr _codec;  // 编解码器
        Consumer::ptr _consumer;  // 消费者
        std::mutex _mutex;
        std::condition_variable _cv;
        // 该map用于存储未接收到服务器返回响应请求的rid和响应的对象
        std::unordered_map<std::string, basicCommonReponsePtr> _basic_resp;  // <rid, 请求对象> -- 基本响应映射表
    };

    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
    public:
        ChannelManager() = default;
        ~ChannelManager() = default;
    public:
        Channel::ptr create(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec)
        {
            std::lock_guard<std::mutex> lock(_mutex);
            auto channel = std::make_shared<Channel>(conn, codec);
            const std::string cid = channel->cid();
            _channels.insert(std::make_pair(cid, channel));
            return channel;
        }

        void remove(const std::string& cid)
        {
            std::lock_guard<std::mutex> lock(_mutex);
            _channels.erase(cid);
        }

        Channel::ptr get(const std::string& cid)
        {
            std::lock_guard<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if(it == _channels.end())
            {
                return nullptr;
            }
            return it->second;
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
}


